package com.lagou.bak;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class WaterMarkTest {

    public static void main(String[] args) {
        /**
         * 1、获取数据源
         * 2、转化成Tuple2<String, Long>
         * 3、watermark ： 等等延迟的数据
         * 4、keyby分组
         * 5、Window机制
         * 6、触发窗口计算
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(10);


        //1、获取数据源
        String hostname = "hdp-1";
        int port = 7777;
        DataStreamSource<String> data = env.socketTextStream(hostname, port);

        //2、转化成Tuple2<String, Long>
        SingleOutputStreamOperator<Tuple2<String, Long>> maped = data.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String value) throws Exception {
                //value:01,1586489566000
                String[] split = value.split(",");

                return new Tuple2<String, Long>(split[0], Long.valueOf(split[1]));
            }
        });

        //3、watermark ： 等等延迟的数据
        SingleOutputStreamOperator<Tuple2<String, Long>> watermarked = maped.assignTimestampsAndWatermarks(new WatermarkStrategy<Tuple2<String, Long>>() {
            @Override
            public WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Tuple2<String, Long>>() {
                    Long maxtimestamp = 0L;
                    Long timeoutoforder = 1000L;

                    @Override //每来一条数据，执行一次的方法
                    public void onEvent(Tuple2<String, Long> event, long eventTimestamp, WatermarkOutput output) {
                        maxtimestamp = Math.max(maxtimestamp, event.f1);
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxtimestamp - timeoutoforder));
                    }
                };
            }
        }.withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                return element.f1;
            }
        }));

        //4、keyby分组
        KeyedStream<Tuple2<String, Long>, String> keyed = watermarked.keyBy(new KeySelector<Tuple2<String, Long>, String>() {
            @Override
            public String getKey(Tuple2<String, Long> value) throws Exception {
                return value.f0;
            }
        });
        //5、Window机制
        WindowedStream<Tuple2<String, Long>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        //6、触发窗口计算
        SingleOutputStreamOperator<String> applyed = windowed.apply(new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
            /**
             *
             * @param s key  01
             * @param window 窗口： 窗口的开始时间，结束时间， 一共有多少元素
             * @param input: 窗口内的所有的数据
             * @param out
             * @throws Exception
             */
            @Override
            public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<String> out) throws Exception {
                out.collect("key:" + s + "...winStart:" + window.getStart() + "...winEnd:" + window.getEnd());
            }
        });

        applyed.print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}



























